/*-------------------------------------------------------------------------
 *
 * syncrep.c
 *
 * 从 PostgreSQL 9.1 开始，异步复制是新的。
 *
 * 如果请求，事务提交会等待其提交 LSN 被同步备用确认。
 *
 * 此模块包含等待和释放后端的代码。
 * 本模块中的所有代码在主服务器上执行。核心流复制
 * 传输保持在 WALreceiver/WALsender 模块内。
 *
 * 该设计的本质是将所有等待/释放的逻辑隔离到主服务器上。
 * 主服务器定义它希望等待的备用。备用完全不知
 * 主服务器上的事务的持久性要求，从而减少代码复杂性并
 * 精简备用操作和网络带宽，因为不需要传递每个事务的状态信息。
 *
 * 复制可以是同步的，也可以是非同步的（异步）。如果是
 * 异步的，我们就快速返回。如果是同步的，那么我们会
 * 在释放等待的后端之前等待备用上的写入、刷新或应用位置。
 * 在以后的版本中，预计这种交互会更复杂。
 *
 * 管理等待后端的最佳方式是拥有一个唯一的等待后端
 * 排序队列，这样我们就可以避免每次接收到回复时都
 * 需要搜索所有等待者。
 *
 * 在 9.5 或更早版本中，只有一个备用可以被视为
 * 同步的。在 9.6 中，我们支持基于优先级的多个同步
 * 备用。在 10.0中也支持基于法定人数的多个同步备用。
 * 事务必须等待回复的同步备用的数量在同步备用名称
 * 中指定。该参数还指定备用名称的列表和从列出名称中选择
 * 同步备用的方法（FIRST 和 ANY）。
 *
 * 方法 FIRST 指定基于优先级的同步复制
 * 并使事务提交等待，直到其 WAL 记录被复制到根据其优
 * 先级选择的请求数量的同步备用。名称较早出现在列表
 * 中的备用获得更高的优先级，并被视为同步。
 * 在副本中后续出现的备用服务器表示潜在的同步备用。
 * 如果当前的任何同步备用由于某种原因断开，它将立即被
 * 具有下一个最高优先级的备用替换。
 *
 * 方法 ANY 指定基于法定人数的同步复制
 * 并使事务提交等待，直到其 WAL 记录被复制到列表中
 * 至少请求的数量的同步备用。所有列出中的备用被视为
 * 法定人数同步备用的候选者。
 *
 * 如果既没有指定 FIRST 也没有指定 ANY，将使用 FIRST
 * 作为方法。这是为了与 9.6 或更早版本的向后兼容，
 * 那里只支持基于优先级的同步复制。
 *
 * 在从同步备用名称中选择的备用可以成为同步备用之前，
 * 它们必须与主服务器保持同步；这可能需要一些时间。
 * 一旦赶上，被视为同步的备用将在那一刻释放队列中的等待者。
 *
 * Portions Copyright (c) 2010-2022, PostgreSQL Global Development Group
 *
 * IDENTIFICATION
 *	  src/backend/replication/syncrep.c
 *
 *-------------------------------------------------------------------------
 */
#include "postgres.h"

#include <unistd.h>

#include "access/xact.h"
#include "miscadmin.h"
#include "pgstat.h"
#include "replication/syncrep.h"
#include "replication/walsender.h"
#include "replication/walsender_private.h"
#include "storage/pmsignal.h"
#include "storage/proc.h"
#include "tcop/tcopprot.h"
#include "utils/builtins.h"
#include "utils/ps_status.h"

#ifdef FDD
#include "replication/logical.h"
#endif
/* 用户可设置的同步复制参数 */
char	   *SyncRepStandbyNames;

#define SyncStandbysDefined() \
	(SyncRepStandbyNames != NULL && SyncRepStandbyNames[0] != '\0')

static bool announce_next_takeover = true;

SyncRepConfigData *SyncRepConfig = NULL;
static int	SyncRepWaitMode = SYNC_REP_NO_WAIT;

static void fc_SyncRepQueueInsert(int fc_mode);
static void fc_SyncRepCancelWait(void);
static int	fc_SyncRepWakeQueue(bool fc_all, int fc_mode);

static bool fc_SyncRepGetSyncRecPtr(XLogRecPtr *fc_writePtr,
								 XLogRecPtr *fc_flushPtr,
								 XLogRecPtr *fc_applyPtr,
								 bool *fc_am_sync);
static void fc_SyncRepGetOldestSyncRecPtr(XLogRecPtr *fc_writePtr,
									   XLogRecPtr *fc_flushPtr,
									   XLogRecPtr *fc_applyPtr,
									   SyncRepStandbyData *fc_sync_standbys,
									   int fc_num_standbys);
static void fc_SyncRepGetNthLatestSyncRecPtr(XLogRecPtr *fc_writePtr,
										  XLogRecPtr *fc_flushPtr,
										  XLogRecPtr *fc_applyPtr,
										  SyncRepStandbyData *fc_sync_standbys,
										  int fc_num_standbys,
										  uint8 fc_nth);
static int	fc_SyncRepGetStandbyPriority(void);
static int	fc_standby_priority_comparator(const void *fc_a, const void *fc_b);
static int	fc_cmp_lsn(const void *fc_a, const void *fc_b);

#ifdef USE_ASSERT_CHECKING
static bool fc_SyncRepQueueIsOrderedByLSN(int fc_mode);
#endif

/*
 * ===========================================================
 * 正常用户后端的同步复制函数
 * ===========================================================
 */

/*
 * 如果用户要求，等待同步复制。
 *
 * 最初后端以状态 SYNC_REP_NOT_WAITING 开始，然后在将自己
 * 添加到等待队列之前将状态改为 SYNC_REP_WAITING。在 SyncRepWakeQueue() 中，
 * 一个 WALSender 在确认复制后将状态更改为 SYNC_REP_WAIT_COMPLETE。
 * 然后，该后端将状态重置为 SYNC_REP_NOT_WAITING。
 *
 * 'lsn' 表示要等待的 LSN。'commit' 指示此 LSN
 * 是否表示提交记录。如果不是，则仅在 synchronous_commit 设置为更高级别的
 * remote_apply 时，我们才会等待 WAL 被刷新，因为只有提交记录提供应用反馈。
 */
void SyncRepWaitForLSN(XLogRecPtr fc_lsn, bool fc_commit)
{
	char	   *fc_new_status = NULL;
	const char *fc_old_status;
	int			fc_mode;

	/*
	 * 此函数应在事务提交期间持有中断时调用，
	 * 以防止后续的共享内存队列清理受到外部中断的影响。
	 */
	Assert(InterruptHoldoffCount > 0);

	/*
	 * 如果用户没有请求同步复制，或没有定义
	 * 同步复制备用名称，则快速退出。
	 *
	 * 由于该例程在每次提交时都会被调用，因此如果未请求同步复制，
	 * 快速退出非常重要。所以我们在不加锁的情况下检查
	 * WalSndCtl->sync_standbys_defined 标志，如果为 false 则立即退出。
	 * 如果为 true，则需要在持有锁的情况下再次检查，以原子方式检查标志
	 * 并操作同步复制队列。这是为了避免在 SyncRepUpdateSyncStandbysDefined()
	 * 中描述的竞争条件。另一方面，如果为 false，则不需要加锁，
	 * 因为我们不会触及队列。
	 */
	if (!SyncRepRequested() ||
		!((volatile WalSndCtlData *) WalSndCtl)->sync_standbys_defined)
		return;

	/* 将其他非提交级别限制为仅远程刷新。 */
	if (fc_commit)
		fc_mode = SyncRepWaitMode;
	else
		fc_mode = Min(SyncRepWaitMode, SYNC_REP_WAIT_FLUSH);

	Assert(SHMQueueIsDetached(&(MyProc->syncRepLinks)));
	Assert(WalSndCtl != NULL);

	LWLockAcquire(SyncRepLock, LW_EXCLUSIVE);
	Assert(MyProc->syncRepState == SYNC_REP_NOT_WAITING);

	/*
	 * 如果 WalSndCtl->sync_standbys_defined 未设置，
	 * 我们不等待同步复制。参见 SyncRepUpdateSyncStandbysDefined。
	 *
	 * 还检查备用是否已经回复。虽然竞争条件不太可能，但我们无论如何都会获取该缓存行，
	 * 所以这个检查可能成本很低。
	 */
	if (!WalSndCtl->sync_standbys_defined ||
		fc_lsn <= WalSndCtl->lsn[fc_mode])
	{
		LWLockRelease(SyncRepLock);
		return;
	}

	/*
	 * 设置我们的 waitLSN，以便 WALSender 知道何时唤醒我们，并将
	 * 自己添加到队列中。
	 */
	MyProc->waitLSN = fc_lsn;
	MyProc->syncRepState = SYNC_REP_WAITING;
	fc_SyncRepQueueInsert(fc_mode);
	Assert(fc_SyncRepQueueIsOrderedByLSN(fc_mode));
	LWLockRelease(SyncRepLock);

	/* 修改 ps 显示以显示正在等待同步复制。 */
	if (update_process_title)
	{
		int			fc_len;

		fc_old_status = get_ps_display(&fc_len);
		fc_new_status = (char *) palloc(fc_len + 32 + 1);
		memcpy(fc_new_status, fc_old_status, fc_len);
		sprintf(fc_new_status + fc_len, " waiting for %X/%X",
				LSN_FORMAT_ARGS(fc_lsn));
		set_ps_display(fc_new_status);
		fc_new_status[fc_len] = '\0'; /* 剪切掉 " waiting ..." */
	}

	/*
	 * 等待指定的 LSN 被确认。
	 *
	 * 每个进程都有自己的等待闩锁，因此我们在这里执行正常的闩锁
	 * 检查/等待循环。
	 */
	for (;;)
	{
		int			fc_rc;

		/* 在测试状态之前必须重置闩锁。 */
		ResetLatch(MyLatch);

		/*
		 * 获取锁是没有必要的，闩锁确保了适当的屏障。
		 * 如果看起来我们完成了，我们必须确实完成，
		 * 因为一旦 walsender 将状态更改为 SYNC_REP_WAIT_COMPLETE，
		 * 它将永远不会再次更新，因此在这种情况下我们不能看到过期值。
		 */
		if (MyProc->syncRepState == SYNC_REP_WAIT_COMPLETE)
			break;

		/*
		 * 如果等待同步复制处于挂起状态，我们既不能
		 * 确认提交，也不能引发 ERROR 或 FATAL。后者会导致
		 * 客户端相信事务已中止，这实际上并不真实：它已经
		 * 在本地提交。前者也不合适：客户端请求了同步
		 * 复制，可以合理地假设已确认的提交也是被复制的，
		 * 这可能并不真实。所以在这种情况下，我们发出一个
		 * WARNING（某些客户端可能能够理解）并停止进一步输出。
		 * 我们不会重置 ProcDiePending，以便该进程在
		 * 清理完提交后死亡。
		 */
		if (ProcDiePending)
		{
			ereport(WARNING,
					(errcode(ERRCODE_ADMIN_SHUTDOWN),
					 errmsg("canceling the wait for synchronous replication and terminating connection due to administrator command"),
					 errdetail("The transaction has already committed locally, but might not have been replicated to the standby.")));
			whereToSendOutput = DestNone;
			fc_SyncRepCancelWait();
			break;
		}

		/*
		 * 如果查询取消中断到达，我们该怎么办还不清楚。
		 * 此时我们实际上无法中止，但完全忽略中断也没有帮助，
		 * 所以我们只是用适当的警告终止等待。
		 */
		if (QueryCancelPending)
		{
			QueryCancelPending = false;
			ereport(WARNING,
					(errmsg("canceling wait for synchronous replication due to user request"),
					 errdetail("The transaction has already committed locally, but might not have been replicated to the standby.")));
			fc_SyncRepCancelWait();
			break;
		}

		/*
		 * 等待锁。任何应该唤醒我们的条件都会设置锁，
		 * 所以不需要超时。
		 */
		fc_rc = WaitLatch(MyLatch, WL_LATCH_SET | WL_POSTMASTER_DEATH, -1,
					   WAIT_EVENT_SYNC_REP);

		/*
		 * 如果主进程死亡，我们可能永远不会收到确认，
		 * 因为所有的 wal sender 进程都会退出。所以干脆退出。
		 */
		if (fc_rc & WL_POSTMASTER_DEATH)
		{
			ProcDiePending = true;
			whereToSendOutput = DestNone;
			fc_SyncRepCancelWait();
			break;
		}
	}

	/*
	 * WalSender 已检查我们的 LSN 并已将我们从队列中移除。
	 * 清理状态并离开。重置这些共享内存字段不需要持有
	 * SyncRepLock，因为不在队列中的任何 wal sender
	 * 都会忽略我们。我们需要一个读取屏障以确保我们看到
	 * 队列链接的变化（如果没有断言，这可能是多余的，
	 * 但未雨绸缪总比事后懊悔好）。
	 */
	pg_read_barrier();
	Assert(SHMQueueIsDetached(&(MyProc->syncRepLinks)));
	MyProc->syncRepState = SYNC_REP_NOT_WAITING;
	MyProc->waitLSN = 0;

	if (fc_new_status)
	{
		/* 重置 ps 显示 */
		set_ps_display(fc_new_status);
		pfree(fc_new_status);
	}
}

/*
 * 将 MyProc 插入指定的 SyncRepQueue，保持排序不变。
 *
 * 通常我们会在队列尾部插入，尽管我们有可能是
 * 超顺序到达这里，所以从尾部开始并反向到插入点。
 */
static void fc_SyncRepQueueInsert(int fc_mode)
{
	PGPROC	   *fc_proc;

	Assert(fc_mode >= 0 && fc_mode < NUM_SYNC_REP_WAIT_MODE);
	fc_proc = (PGPROC *) SHMQueuePrev(&(WalSndCtl->SyncRepQueue[fc_mode]),
								   &(WalSndCtl->SyncRepQueue[fc_mode]),
								   offsetof(PGPROC, syncRepLinks));

	while (fc_proc)
	{
		/*
		 * 停止在我们应该到达的队列元素，确保队列
		 * 按 LSN 排序。
		 */
		if (fc_proc->waitLSN < MyProc->waitLSN)
			break;

		fc_proc = (PGPROC *) SHMQueuePrev(&(WalSndCtl->SyncRepQueue[fc_mode]),
									   &(fc_proc->syncRepLinks),
									   offsetof(PGPROC, syncRepLinks));
	}

	if (fc_proc)
		SHMQueueInsertAfter(&(fc_proc->syncRepLinks), &(MyProc->syncRepLinks));
	else
		SHMQueueInsertAfter(&(WalSndCtl->SyncRepQueue[fc_mode]), &(MyProc->syncRepLinks));
}

/*
 * 获取 SyncRepLock 并取消当前正在进行的任何等待。
 */
static void fc_SyncRepCancelWait(void)
{
	LWLockAcquire(SyncRepLock, LW_EXCLUSIVE);
	if (!SHMQueueIsDetached(&(MyProc->syncRepLinks)))
		SHMQueueDelete(&(MyProc->syncRepLinks));
	MyProc->syncRepState = SYNC_REP_NOT_WAITING;
	LWLockRelease(SyncRepLock);
}

void SyncRepCleanupAtProcExit(void)
{
	/*
	 * 首先检查我们是否在没有锁的情况下从队列中删除，以免
	 * 减慢后端退出。
	 */
	if (!SHMQueueIsDetached(&(MyProc->syncRepLinks)))
	{
		LWLockAcquire(SyncRepLock, LW_EXCLUSIVE);

		/* 也许我们刚被移除，所以重新检查 */
		if (!SHMQueueIsDetached(&(MyProc->syncRepLinks)))
			SHMQueueDelete(&(MyProc->syncRepLinks));

		LWLockRelease(SyncRepLock);
	}
}

/*
 * ===========================================================
 * wal sender 进程的同步复制函数
 * ===========================================================
 */

/*
 * 采取任何必要的行动以根据配置数据初始化同步复制状态。
 * 在 WALSender 启动和每次 SIGHUP 后调用。
 */
#ifdef FDD
void SyncRepInitConfig(bool fc_failover_slot)
#else
void SyncRepInitConfig(void)
#endif
{
	int			fc_priority;

	/*
	 * 确定我们是否是潜在的同步备用，并记住结果以
	 * 处理来自备用的回复。
	 */
#ifdef FDD
	if (fc_failover_slot)
	{
		//可故障倒换逻辑复制槽改为异步
		fc_priority = 0;
	}
	else
#endif
	fc_priority = fc_SyncRepGetStandbyPriority();
	if (MyWalSnd->sync_standby_priority != fc_priority)
	{
		SpinLockAcquire(&MyWalSnd->mutex);
		MyWalSnd->sync_standby_priority = fc_priority;
		SpinLockRelease(&MyWalSnd->mutex);

		ereport(DEBUG1,
				(errmsg_internal("standby \"%s\" now has synchronous standby priority %u",
								 application_name, fc_priority)));
	}
}

/*
 * 根据我们最新的状态更新每个队列中的 LSN。
 * 这实现了一个简单的策略，即第一个有效同步备用释放等待者。
 *
 * 其他策略也是可能的，这将改变我们在这里所做的事情
 * 以及我们可能存储哪些信息。
 */
void SyncRepReleaseWaiters(void)
{
	volatile WalSndCtlData *fc_walsndctl = WalSndCtl;
	XLogRecPtr	fc_writePtr;
	XLogRecPtr	fc_flushPtr;
	XLogRecPtr	fc_applyPtr;
	bool		fc_got_recptr;
	bool		fc_am_sync;
	int			fc_numwrite = 0;
	int			fc_numflush = 0;
	int			fc_numapply = 0;

	/*
	 * 如果这个 WALSender 正在为一个不在潜在同步备用列表上的备用服务，那么我们没有任何事情要做。如果我们仍在启动，仍在运行基本备份，或者当前的刷新位置仍然无效，那么也应该迅速离开。流式传输或停止 WAL 发送者可以释放等待者。
	 */
	if (MyWalSnd->sync_standby_priority == 0 ||
		(MyWalSnd->state != WALSNDSTATE_STREAMING &&
		 MyWalSnd->state != WALSNDSTATE_STOPPING) ||
		XLogRecPtrIsInvalid(MyWalSnd->flush))
	{
		announce_next_takeover = true;
		return;
	}

	/*
	 * 我们是潜在的同步备用。如果有足够的同步备用并且我们被认为是同步的，则释放等待者。
	 */
	LWLockAcquire(SyncRepLock, LW_EXCLUSIVE);

	/*
	 * 检查我们是否是同步备用，并计算所有同步备用之间的同步位置。（注意：尽管这一步本身不需要保持 SyncRepLock，但在获得锁之后进行此操作似乎是个不错的主意。这确保我们用来释放等待者的 WAL 指针比之前的任何执行此例程的执行都要新。）
	 */
	fc_got_recptr = fc_SyncRepGetSyncRecPtr(&fc_writePtr, &fc_flushPtr, &fc_applyPtr, &fc_am_sync);

	/*
	 * 如果我们正在管理一个同步备用，尽管之前不是，那么宣布我们现在是一个同步备用。
	 */
	if (announce_next_takeover && fc_am_sync)
	{
		announce_next_takeover = false;

		if (SyncRepConfig->syncrep_method == SYNC_REP_PRIORITY)
			ereport(LOG,
					(errmsg("standby \"%s\" is now a synchronous standby with priority %u",
							application_name, MyWalSnd->sync_standby_priority)));
		else
			ereport(LOG,
					(errmsg("standby \"%s\" is now a candidate for quorum synchronous standby",
							application_name)));
	}

	/*
	 * 如果同步备用的数量少于请求或者我们没有管理同步备用，那么就离开。
	 */
	if (!fc_got_recptr || !fc_am_sync)
	{
		LWLockRelease(SyncRepLock);
		announce_next_takeover = !fc_am_sync;
		return;
	}

	/*
	 * 首先设置 lsn，以便当我们唤醒后端时，它们将释放到这个位置。
	 */
	if (fc_walsndctl->lsn[SYNC_REP_WAIT_WRITE] < fc_writePtr)
	{
		fc_walsndctl->lsn[SYNC_REP_WAIT_WRITE] = fc_writePtr;
		fc_numwrite = fc_SyncRepWakeQueue(false, SYNC_REP_WAIT_WRITE);
	}
	if (fc_walsndctl->lsn[SYNC_REP_WAIT_FLUSH] < fc_flushPtr)
	{
		fc_walsndctl->lsn[SYNC_REP_WAIT_FLUSH] = fc_flushPtr;
		fc_numflush = fc_SyncRepWakeQueue(false, SYNC_REP_WAIT_FLUSH);
	}
	if (fc_walsndctl->lsn[SYNC_REP_WAIT_APPLY] < fc_applyPtr)
	{
		fc_walsndctl->lsn[SYNC_REP_WAIT_APPLY] = fc_applyPtr;
		fc_numapply = fc_SyncRepWakeQueue(false, SYNC_REP_WAIT_APPLY);
	}

	LWLockRelease(SyncRepLock);

	elog(DEBUG3, "released %d procs up to write %X/%X, %d procs up to flush %X/%X, %d procs up to apply %X/%X",
		 fc_numwrite, LSN_FORMAT_ARGS(fc_writePtr),
		 fc_numflush, LSN_FORMAT_ARGS(fc_flushPtr),
		 fc_numapply, LSN_FORMAT_ARGS(fc_applyPtr));
}

/*
 * 计算同步备用之间的同步写入、刷新和应用位置。
 *
 * 如果同步备用的数量少于 synchronous_standby_names 指定的数量，则返回 false。否则返回 true，并将位置存储到 *writePtr、*flushPtr 和 *applyPtr。
 *
 * 返回时，如果该 walsender 正在连接到同步备用，则 *am_sync 被设置为 true。否则设置为 false。
 */
static bool fc_SyncRepGetSyncRecPtr(XLogRecPtr *fc_writePtr, XLogRecPtr *fc_flushPtr,
					 XLogRecPtr *fc_applyPtr, bool *fc_am_sync)
{
	SyncRepStandbyData *fc_sync_standbys;
	int			fc_num_standbys;
	int			fc_i;

	/* 初始化默认结果 */
	*fc_writePtr = InvalidXLogRecPtr;
	*fc_flushPtr = InvalidXLogRecPtr;
	*fc_applyPtr = InvalidXLogRecPtr;
	*fc_am_sync = false;

	/* 如果连配置为同步都没有，就快速退出 */
	if (SyncRepConfig == NULL)
		return false;

	/* 获取此刻被认为是同步的备用 */
	fc_num_standbys = SyncRepGetCandidateStandbys(&fc_sync_standbys);

	/* 我是否在候选同步备用中？ */
	for (fc_i = 0; fc_i < fc_num_standbys; fc_i++)
	{
		if (fc_sync_standbys[fc_i].is_me)
		{
			*fc_am_sync = true;
			break;
		}
	}

	/*
	 * 如果我们没有管理一个同步备用或没有足够的同步备用，那么就没有更多的事情要做。
	 */
	if (!(*fc_am_sync) ||
		fc_num_standbys < SyncRepConfig->num_sync)
	{
		pfree(fc_sync_standbys);
		return false;
	}

	/*
	 * 在基于优先级的同步复制中，同步的写入位置是同步备用中最旧的。在基于法定人数的情况下，它们是第 N 个最新的位置。
	 *
	 * SyncRepGetNthLatestSyncRecPtr() 也可以计算最旧的位置。但我们使用 SyncRepGetOldestSyncRecPtr() 进行该计算，因为它效率更高一些。
	 *
	 * XXX 如果当前和请求的同步备用数量相同，我们可以即使在基于法定人数的同步复制中使用 SyncRepGetOldestSyncRecPtr() 计算同步位置。
	 */
	if (SyncRepConfig->syncrep_method == SYNC_REP_PRIORITY)
	{
		fc_SyncRepGetOldestSyncRecPtr(fc_writePtr, fc_flushPtr, fc_applyPtr,
								   fc_sync_standbys, fc_num_standbys);
	}
	else
	{
		fc_SyncRepGetNthLatestSyncRecPtr(fc_writePtr, fc_flushPtr, fc_applyPtr,
									  fc_sync_standbys, fc_num_standbys,
									  SyncRepConfig->num_sync);
	}

	pfree(fc_sync_standbys);
	return true;
}

/*
 * 计算同步备用中的最旧写入、刷新和应用位置。
 */
static void fc_SyncRepGetOldestSyncRecPtr(XLogRecPtr *fc_writePtr,
						   XLogRecPtr *fc_flushPtr,
						   XLogRecPtr *fc_applyPtr,
						   SyncRepStandbyData *fc_sync_standbys,
						   int fc_num_standbys)
{
	int			fc_i;

	/*
	 * 扫描所有同步备用并计算最旧的写入、刷新和应用位置。我们假设 *writePtr 等被初始化为 InvalidXLogRecPtr。
	 */
	for (fc_i = 0; fc_i < fc_num_standbys; fc_i++)
	{
		XLogRecPtr	fc_write = fc_sync_standbys[fc_i].write;
		XLogRecPtr	fc_flush = fc_sync_standbys[fc_i].flush;
		XLogRecPtr	fc_apply = fc_sync_standbys[fc_i].apply;

		if (XLogRecPtrIsInvalid(*fc_writePtr) || *fc_writePtr > fc_write)
			*fc_writePtr = fc_write;
		if (XLogRecPtrIsInvalid(*fc_flushPtr) || *fc_flushPtr > fc_flush)
			*fc_flushPtr = fc_flush;
		if (XLogRecPtrIsInvalid(*fc_applyPtr) || *fc_applyPtr > fc_apply)
			*fc_applyPtr = fc_apply;
	}
}

/*
 * 计算同步备用中第 N 个最新的写入、刷新和应用位置。
 */
static void fc_SyncRepGetNthLatestSyncRecPtr(XLogRecPtr *fc_writePtr,
							  XLogRecPtr *fc_flushPtr,
							  XLogRecPtr *fc_applyPtr,
							  SyncRepStandbyData *fc_sync_standbys,
							  int fc_num_standbys,
							  uint8 fc_nth)
{
	XLogRecPtr *fc_write_array;
	XLogRecPtr *fc_flush_array;
	XLogRecPtr *fc_apply_array;
	int			fc_i;

	/* 应该有足够的候选者，否则就是有人搞砸了 */
	Assert(fc_nth > 0 && fc_nth <= fc_num_standbys);

	fc_write_array = (XLogRecPtr *) palloc(sizeof(XLogRecPtr) * fc_num_standbys);
	fc_flush_array = (XLogRecPtr *) palloc(sizeof(XLogRecPtr) * fc_num_standbys);
	fc_apply_array = (XLogRecPtr *) palloc(sizeof(XLogRecPtr) * fc_num_standbys);

	for (fc_i = 0; fc_i < fc_num_standbys; fc_i++)
	{
		fc_write_array[fc_i] = fc_sync_standbys[fc_i].write;
		fc_flush_array[fc_i] = fc_sync_standbys[fc_i].flush;
		fc_apply_array[fc_i] = fc_sync_standbys[fc_i].apply;
	}

	/* 将每个数组按降序排序 */
	qsort(fc_write_array, fc_num_standbys, sizeof(XLogRecPtr), fc_cmp_lsn);
	qsort(fc_flush_array, fc_num_standbys, sizeof(XLogRecPtr), fc_cmp_lsn);
	qsort(fc_apply_array, fc_num_standbys, sizeof(XLogRecPtr), fc_cmp_lsn);

	/* 获取最新的 N 次写入、刷新、应用位置 */
	*fc_writePtr = fc_write_array[fc_nth - 1];
	*fc_flushPtr = fc_flush_array[fc_nth - 1];
	*fc_applyPtr = fc_apply_array[fc_nth - 1];

	pfree(fc_write_array);
	pfree(fc_flush_array);
	pfree(fc_apply_array);
}

/*
 * 比较 lsn 以按降序排序数组。
 */
static int fc_cmp_lsn(const void *fc_a, const void *fc_b)
{
	XLogRecPtr	fc_lsn1 = *((const XLogRecPtr *) fc_a);
	XLogRecPtr	fc_lsn2 = *((const XLogRecPtr *) fc_b);

	if (fc_lsn1 > fc_lsn2)
		return -1;
	else if (fc_lsn1 == fc_lsn2)
		return 0;
	else
		return 1;
}

/*
 * 返回关于 walsenders 的数据，这些 walsenders 是同步备用的候选者。
 *
 * *standbys 被设置为一个 palloc'd 数组，包含每个 walsender 数据的结构，
 * 返回有效条目的数量（候选同步发送者）。
 * （这可能多于或少于 num_sync；调用者必须检查。）
 */
int SyncRepGetCandidateStandbys(SyncRepStandbyData **fc_standbys)
{
	int			fc_i;
	int			fc_n;

	/* 创建结果数组 */
	*fc_standbys = (SyncRepStandbyData *)
		palloc(max_wal_senders * sizeof(SyncRepStandbyData));

	/* 如果未请求同步复制，则快速退出 */
	if (SyncRepConfig == NULL)
		return 0;

	/* 从共享内存收集原始数据 */
	fc_n = 0;
	for (fc_i = 0; fc_i < max_wal_senders; fc_i++)
	{
		volatile WalSnd *fc_walsnd;	/* 使用 volatile 指针防止代码
									 * 重排 */
		SyncRepStandbyData *fc_stby;
		WalSndState fc_state;		/* 不包含在 SyncRepStandbyData 中 */

		fc_walsnd = &WalSndCtl->walsnds[fc_i];
		fc_stby = *fc_standbys + fc_n;

		SpinLockAcquire(&fc_walsnd->mutex);
		fc_stby->pid = fc_walsnd->pid;
		fc_state = fc_walsnd->state;
		fc_stby->write = fc_walsnd->write;
		fc_stby->flush = fc_walsnd->flush;
		fc_stby->apply = fc_walsnd->apply;
		fc_stby->sync_standby_priority = fc_walsnd->sync_standby_priority;
		SpinLockRelease(&fc_walsnd->mutex);

		/* 必须处于活动状态 */
		if (fc_stby->pid == 0)
			continue;

		/* 必须正在流或停止 */
		if (fc_state != WALSNDSTATE_STREAMING &&
			fc_state != WALSNDSTATE_STOPPING)
			continue;

		/* 必须是同步的 */
		if (fc_stby->sync_standby_priority == 0)
			continue;

		/* 必须具有有效的刷新位置 */
		if (XLogRecPtrIsInvalid(fc_stby->flush))
			continue;

		/* 好的，它是一个候选者 */
		fc_stby->walsnd_index = fc_i;
		fc_stby->is_me = (fc_walsnd == MyWalSnd);
		fc_n++;
	}

	/*
	 * 在法定人数模式下，我们返回所有候选者。在优先级模式下，如果候选者过多，
	 * 则仅返回最高优先级的 num_sync 个。
	 */
	if (SyncRepConfig->syncrep_method == SYNC_REP_PRIORITY &&
		fc_n > SyncRepConfig->num_sync)
	{
		/* 按优先级排序... */
		qsort(*fc_standbys, fc_n, sizeof(SyncRepStandbyData),
			  fc_standby_priority_comparator);
		/* ... 然后仅报告前 num_sync 个 */
		fc_n = SyncRepConfig->num_sync;
	}

	return fc_n;
}

/*
 * qsort 比较器用于按优先级排序 SyncRepStandbyData 条目
 */
static int fc_standby_priority_comparator(const void *fc_a, const void *fc_b)
{
	const SyncRepStandbyData *fc_sa = (const SyncRepStandbyData *) fc_a;
	const SyncRepStandbyData *fc_sb = (const SyncRepStandbyData *) fc_b;

	/* 首先，按递增的优先级值排序 */
	if (fc_sa->sync_standby_priority != fc_sb->sync_standby_priority)
		return fc_sa->sync_standby_priority - fc_sb->sync_standby_priority;

	/*
	 * 我们可能有相等的优先级值；通过在 WALSnd 数组中的位置任意打破平局。
	 * （这完全是错误的，因为这是依赖于到达顺序的，但有回归测试依赖于此。）
	 */
	return fc_sa->walsnd_index - fc_sb->walsnd_index;
}


/*
 * 检查我们是否在同步备用列表中，如果是，确定优先级顺序。
 * 如果设置了优先级，则返回优先级，否则返回零以表明
 * 我们不是潜在的同步备用。
 *
 * 将参数 SyncRepStandbyNames 与此 WALSender 的 application_name 进行比较，
 * 如果发现通配符 "*"，则允许任何名称。
 */
static int fc_SyncRepGetStandbyPriority(void)
{
	const char *fc_standby_name;
	int			fc_priority;
	bool		fc_found = false;

	/*
	 * 由于不允许同步级联复制，我们始终将级联 walsender 的
	 * 优先级设置为零。
	 */
	if (am_cascading_walsender)
		return 0;

	if (!SyncStandbysDefined() || SyncRepConfig == NULL)
		return 0;

	fc_standby_name = SyncRepConfig->member_names;
	for (fc_priority = 1; fc_priority <= SyncRepConfig->nmembers; fc_priority++)
	{
		if (pg_strcasecmp(fc_standby_name, application_name) == 0 ||
			strcmp(fc_standby_name, "*") == 0)
		{
			fc_found = true;
			break;
		}
		fc_standby_name += strlen(fc_standby_name) + 1;
	}

	if (!fc_found)
		return 0;

	/*
	 * 在基于法定人数的同步复制中，列表中的所有备用具有
	 * 相同的优先级，即一。
	 */
	return (SyncRepConfig->syncrep_method == SYNC_REP_PRIORITY) ? fc_priority : 1;
}

/*
 * 从队列头部遍历指定队列。 设置任何需要被唤醒的后端的状态，
 * 将它们从队列中移除，然后唤醒它们。
 * 传入 all = true 可唤醒整个队列；否则，只唤醒到
 * walsender 的 LSN。
 *
 * 调用者必须以独占模式持有 SyncRepLock。
 */
static int fc_SyncRepWakeQueue(bool fc_all, int fc_mode)
{
	volatile WalSndCtlData *fc_walsndctl = WalSndCtl;
	PGPROC	   *fc_proc = NULL;
	PGPROC	   *fc_thisproc = NULL;
	int			fc_numprocs = 0;

	Assert(fc_mode >= 0 && fc_mode < NUM_SYNC_REP_WAIT_MODE);
	Assert(LWLockHeldByMeInMode(SyncRepLock, LW_EXCLUSIVE));
	Assert(fc_SyncRepQueueIsOrderedByLSN(fc_mode));

	fc_proc = (PGPROC *) SHMQueueNext(&(WalSndCtl->SyncRepQueue[fc_mode]),
								   &(WalSndCtl->SyncRepQueue[fc_mode]),
								   offsetof(PGPROC, syncRepLinks));

	while (fc_proc)
	{
		/*
		 * 假设队列是按 LSN 排序的
		 */
		if (!fc_all && fc_walsndctl->lsn[fc_mode] < fc_proc->waitLSN)
			return fc_numprocs;

		/*
		 * 移动到下一个进程，以便我们可以从队列中删除此进程。
		 * thisproc 是有效的，proc 可能在此后为 NULL。
		 */
		fc_thisproc = fc_proc;
		fc_proc = (PGPROC *) SHMQueueNext(&(WalSndCtl->SyncRepQueue[fc_mode]),
									   &(fc_proc->syncRepLinks),
									   offsetof(PGPROC, syncRepLinks));

		/*
		 * 从队列中移除此进程。
		 */
		SHMQueueDelete(&(fc_thisproc->syncRepLinks));

		/*
		 * SyncRepWaitForLSN() 在未持有锁的情况下读取 syncRepState，
		 * 因此要确保它在 syncRepState 变更之前看到队列链接被移除。
		 */
		pg_write_barrier();

		/*
		 * 将状态设置为完成；有关各种状态的讨论，请参见 SyncRepWaitForLSN()。
		 */
		fc_thisproc->syncRepState = SYNC_REP_WAIT_COMPLETE;

		/*
		 * 只有在我们设置状态并从队列中移除时才唤醒。
		 */
		SetLatch(&(fc_thisproc->procLatch));

		fc_numprocs++;
	}

	return fc_numprocs;
}

/*
 * 检查点在需要时调用此函数来更新共享的
 * sync_standbys_defined 标志，以便后端不会在
 * synchronous_standby_names 未设置的情况下永远保持耦合。
 * 检查当前值时无需加锁，因为它只会由一个进程更新。
 * 但是更改它时必须加锁。
 */
void SyncRepUpdateSyncStandbysDefined(void)
{
	bool		fc_sync_standbys_defined = SyncStandbysDefined();

	if (fc_sync_standbys_defined != WalSndCtl->sync_standbys_defined)
	{
		LWLockAcquire(SyncRepLock, LW_EXCLUSIVE);

		/*
		 * 如果 synchronous_standby_names 已重置为空，
		 * 后端继续等待是徒劳的。由于用户不再需要
		 * 同步复制，我们最好将它们唤醒。
		 */
		if (!fc_sync_standbys_defined)
		{
			int			fc_i;

			for (fc_i = 0; fc_i < NUM_SYNC_REP_WAIT_MODE; fc_i++)
				fc_SyncRepWakeQueue(true, fc_i);
		}

		/*
		 * 只有在定义了同步备用时，才允许人员加入队列。
		 * 没有这种互锁，可能会出现竞争条件：
		 * 我们可能会唤醒所有当前的等待者；然后，某些
		 * 尚未重新加载其配置的后端可能会在队列中睡眠
		 * （并且永远不会醒来）。这可以防止这种情况发生。
		 */
		WalSndCtl->sync_standbys_defined = fc_sync_standbys_defined;

		LWLockRelease(SyncRepLock);
	}
}

#ifdef USE_ASSERT_CHECKING
static bool fc_SyncRepQueueIsOrderedByLSN(int fc_mode)
{
	PGPROC	   *fc_proc = NULL;
	XLogRecPtr	fc_lastLSN;
#ifdef FDD
	bool fc_lastisNotLogical = true;// 记录上一次是否是非逻辑复制解密进程，注意：若当前比较的lsn和上一次的相等时，只有当前为逻辑解密进程，才会设置为true，否则保持不变
#endif
	Assert(fc_mode >= 0 && fc_mode < NUM_SYNC_REP_WAIT_MODE);

	fc_lastLSN = 0;

	fc_proc = (PGPROC *) SHMQueueNext(&(WalSndCtl->SyncRepQueue[fc_mode]),
								   &(WalSndCtl->SyncRepQueue[fc_mode]),
								   offsetof(PGPROC, syncRepLinks));

	while (fc_proc)
	{
#ifdef FDD
		if(fc_proc->waitLSN == fc_lastLSN)
		{
			if(!(fc_proc->statusFlags & PROC_IN_LOGICAL_DECODING))
			{
				if(fc_lastisNotLogical)
				{
					return false;
				}
				else
				{
					fc_lastisNotLogical = true;
				}
			}                        
		}
		else
		{
			fc_lastisNotLogical = (fc_proc->statusFlags & PROC_IN_LOGICAL_DECODING) ? false : true;
		}
#endif
		/*
		 * 检查队列是否按 LSN 排序，并且多个进程没有
		 * 匹配的 LSN。
		 */
#ifdef FDD
		if (fc_proc->waitLSN < fc_lastLSN)
#else
		if (fc_proc->waitLSN <= fc_lastLSN)
#endif
			return false;

		fc_lastLSN = fc_proc->waitLSN;

		fc_proc = (PGPROC *) SHMQueueNext(&(WalSndCtl->SyncRepQueue[fc_mode]),
									   &(fc_proc->syncRepLinks),
									   offsetof(PGPROC, syncRepLinks));
	}

	return true;
}
#endif

/*
 * ===========================================================
 * 由任何进程执行的同步复制功能
 * ===========================================================
 */

bool check_synchronous_standby_names(char **fc_newval, void **fc_extra, GucSource fc_source)
{
	if (*fc_newval != NULL && (*fc_newval)[0] != '\0')
	{
		int			fc_parse_rc;
		SyncRepConfigData *fc_pconf;

		/* 重置通信变量以确保新开始 */
		syncrep_parse_result = NULL;
		syncrep_parse_error_msg = NULL;

		/* 解析 synchronous_standby_names 字符串 */
		syncrep_scanner_init(*fc_newval);
		fc_parse_rc = syncrep_yyparse();
		syncrep_scanner_finish();

		if (fc_parse_rc != 0 || syncrep_parse_result == NULL)
		{
			GUC_check_errcode(ERRCODE_SYNTAX_ERROR);
			if (syncrep_parse_error_msg)
				GUC_check_errdetail("%s", syncrep_parse_error_msg);
			else
				GUC_check_errdetail("synchronous_standby_names parser failed");
			return false;
		}

		if (syncrep_parse_result->num_sync <= 0)
		{
			GUC_check_errmsg("number of synchronous standbys (%d) must be greater than zero",
							 syncrep_parse_result->num_sync);
			return false;
		}

		/* GUC 额外值必须使用 malloc 分配，而不是 palloc */
		fc_pconf = (SyncRepConfigData *)
			malloc(syncrep_parse_result->config_size);
		if (fc_pconf == NULL)
			return false;
		memcpy(fc_pconf, syncrep_parse_result, syncrep_parse_result->config_size);

		*fc_extra = (void *) fc_pconf;

		/*
		 * 我们无需显式清理 syncrep_parse_result。
		 * 它以及在解析过程中生成的任何其他杂物，都会在
		 * 当前内存上下文被删除时释放。（此代码通常在
		 * 用于配置文件处理的短期上下文中运行，因此
		 * 不会很长。）
		 */
	}
	else
		*fc_extra = NULL;

	return true;
}

void assign_synchronous_standby_names(const char *fc_newval, void *fc_extra)
{
	SyncRepConfig = (SyncRepConfigData *) fc_extra;
}

void assign_synchronous_commit(int fc_newval, void *fc_extra)
{
	switch (fc_newval)
	{
		case SYNCHRONOUS_COMMIT_REMOTE_WRITE:
			SyncRepWaitMode = SYNC_REP_WAIT_WRITE;
			break;
		case SYNCHRONOUS_COMMIT_REMOTE_FLUSH:
			SyncRepWaitMode = SYNC_REP_WAIT_FLUSH;
			break;
		case SYNCHRONOUS_COMMIT_REMOTE_APPLY:
			SyncRepWaitMode = SYNC_REP_WAIT_APPLY;
			break;
		default:
			SyncRepWaitMode = SYNC_REP_NO_WAIT;
			break;
	}
}

#ifdef FDD
//等待LSN在备机处理完成
void FDB_WaitForLsnEndTransaction(XLogRecPtr fc_commitLsn, void *fc_ctx)
{
	//volatile WalSndCtlData *walsndctl = WalSndCtl;
	//XLogRecPtr	comfirmedPtr = InvalidXLogRecPtr;	
	//int			len = 0;
	const char *fc_old_status = NULL;
	char *fc_new_status = NULL;
	int			fc_mode = SyncRepWaitMode;
	
	LogicalDecodingContext *fc_decodeCtx = (LogicalDecodingContext*)fc_ctx;
	//这里是commit，直接使用SyncRepWaitMode对应的队列

	if (fc_mode == SYNC_REP_NO_WAIT)
	{
		//对于failover slot来说，必须有一个同步流复制备，一般来说这是不合理的
		//to do
		return;
	}


	Assert(SHMQueueIsDetached(&(MyProc->syncRepLinks)));
	Assert(WalSndCtl != NULL);

	LWLockAcquire(SyncRepLock, LW_EXCLUSIVE);
	Assert(MyProc->syncRepState == SYNC_REP_NOT_WAITING);

	/*
	 * 如果 WalSndCtl->sync_standbys_defined 未设置，
	 * 我们不等待同步复制。参见 SyncRepUpdateSyncStandbysDefined。
	 *
	 * 还检查备用是否已经回复。虽然竞争条件不太可能，但我们无论如何都会获取该缓存行，
	 * 所以这个检查可能成本很低。
	 */
	if (!WalSndCtl->sync_standbys_defined ||
		fc_commitLsn <= WalSndCtl->lsn[fc_mode])
	{
		LWLockRelease(SyncRepLock);
		return;
	}

	/*
	 * 设置我们的 waitLSN，以便 WALSender 知道何时唤醒我们，并将
	 * 自己添加到队列中。
	 */
	MyProc->waitLSN = fc_commitLsn;
	MyProc->syncRepState = SYNC_REP_WAITING;
	fc_SyncRepQueueInsert(fc_mode);
	Assert(fc_SyncRepQueueIsOrderedByLSN(fc_mode));
	LWLockRelease(SyncRepLock);

	/* 修改 ps 显示以显示正在等待同步复制。 */
	if (update_process_title)
	{
		int			fc_len;

		fc_old_status = get_ps_display(&fc_len);
		fc_new_status = (char *) palloc(fc_len + 32 + 1);
		memcpy(fc_new_status, fc_old_status, fc_len);
		sprintf(fc_new_status + fc_len, " waiting for %X/%X",
				LSN_FORMAT_ARGS(fc_commitLsn));
		set_ps_display(fc_new_status);
		fc_new_status[fc_len] = '\0'; /* 剪切掉 " waiting ..." */
	}

	/*
	 * 等待指定的 LSN 被确认。
	 *
	 * 每个进程都有自己的等待闩锁，因此我们在这里执行正常的闩锁
	 * 检查/等待循环。
	 */
	for (;;)
	{
		/* 在测试状态之前必须重置闩锁。 */
		ResetLatch(MyLatch);

		/*
		 * 获取锁是没有必要的，闩锁确保了适当的屏障。
		 * 如果看起来我们完成了，我们必须确实完成，
		 * 因为一旦 walsender 将状态更改为 SYNC_REP_WAIT_COMPLETE，
		 * 它将永远不会再次更新，因此在这种情况下我们不能看到过期值。
		 */
		if (MyProc->syncRepState == SYNC_REP_WAIT_COMPLETE)
			break;

		if (!fc_decodeCtx->fdb_keepalive_cb())
		{
			//收到stoping
			break;
		}


		//to_do 后续处理没有同步流复制的场景
	}

	/*
	 * WalSender 已检查我们的 LSN 并已将我们从队列中移除。
	 * 清理状态并离开。重置这些共享内存字段不需要持有
	 * SyncRepLock，因为不在队列中的任何 wal sender
	 * 都会忽略我们。我们需要一个读取屏障以确保我们看到
	 * 队列链接的变化（如果没有断言，这可能是多余的，
	 * 但未雨绸缪总比事后懊悔好）。
	 */
	pg_read_barrier();
	Assert(SHMQueueIsDetached(&(MyProc->syncRepLinks)));
	MyProc->syncRepState = SYNC_REP_NOT_WAITING;
	MyProc->waitLSN = 0;

	if (fc_new_status)
	{
		/* 重置 ps 显示 */
		set_ps_display(fc_new_status);
		pfree(fc_new_status);
	}

	SetLatch(MyLatch);
}

#endif
